/**
 * 
 */
package edu.umd.clip.lm.model.data;

import java.io.*;
import java.nio.channels.*;
import java.util.*;

/**
 * @author Denis Filimonov <den@cs.umd.edu>
 *
 */
public class OnDiskTrainingDataNodeReader extends TrainingDataNodeReader implements Observer {
	private ReadableByteChannel input;
	private ReadableTrainingDataNode currentNode;
	private int currentData;
	private TrainingDataReader reader;
	
	public OnDiskTrainingDataNodeReader(ReadableByteChannel input) {
		super();
		this.input = new BufferedInputChannel(input);
		this.reader = new OnDiskTrainingDataReader(this.input);
		this.reader.addFinishedReadingDataObserver(this);
	}

	/* (non-Javadoc)
	 * @see edu.umd.clip.lm.model.training.TrainingDataNodeReader#getNext()
	 */
	@Override
	public ReadableTrainingDataNode getNext() {
		ReadableTrainingDataNode nextItem = null;
		try {
			synchronized(this) {
				nextItem = new ReadableTrainingDataNode(input);
				ReadableTrainingData data[] = new ReadableTrainingData[nextItem.getNumData()];
				for(int i=0; i<data.length; ++i) {
					data[i] = new ReadableTrainingData(reader);
				}
				nextItem.setData(data);
				currentNode = nextItem;
				currentData = 0;
				/*
				System.err.printf("[%s] New current node: %s (%d) %d\n", Thread.currentThread().getName(), 
						currentNode, currentNode.getNodeId(), currentNode.getNumData());
				*/
			}
		} catch(EOFException e) {
			// all finished
			//System.err.println("EOF");
		} catch (IOException e) {
			e.printStackTrace();
		}
		return nextItem;	
	}

	/* (non-Javadoc)
	 * @see java.util.Observer#update(java.util.Observable, java.lang.Object)
	 */
	@Override
	public void update(Observable o, Object arg) {
		ReadableTrainingDataNode oldCurrentNode = null;

		synchronized(this) {
			/*
			System.err.printf("[%s] finished data #%d in %s (%s)\n", Thread.currentThread().getName(), 
					currentData, currentNode, currentNode == null ? "null" : Integer.toString(currentNode.getNodeId()));
			*/
			++currentData;
			if (currentData >= currentNode.getNumData()) {
				currentData = -1;
				oldCurrentNode = currentNode;
				currentNode = null;
			}
		}
		if (oldCurrentNode != null) {
			finishedReadingNode(oldCurrentNode);
		}
	}

	/* (non-Javadoc)
	 * @see edu.umd.clip.lm.model.data.TrainingDataNodeReader#close()
	 */
	@Override
	public void close() throws IOException {
		input.close();
	}

}
